#include "SubReactor.h"
#include "TcpConnection.h"
#include <iostream>
#include <unistd.h>
#include <sys/eventfd.h>
#include <errno.h>
#include <thread>

using std::cout;
using std::endl;
using std::cerr;

SubReactor::SubReactor()
: _epfd(createEpollFd())
, _evtList(1024)
, _isLooping(false)
, _evtfd(createEventFd())
{
    // cout << "[SubReactor] 构造函数: epfd=" << _epfd << ", evtfd=" << _evtfd << endl;
    addEpollReadFd(_evtfd);
    // cout << "[SubReactor] 构造函数完成，已将eventfd加入epoll监听" << endl;
}

SubReactor::~SubReactor()
{
    close(_epfd);
    close(_evtfd);
}

void SubReactor::start()
{
    _subThread = std::thread(&SubReactor::loop, this);
}

void SubReactor::stop()
{
    unloop();
    if (_subThread.joinable()) {
        _subThread.join();
    }
}

void SubReactor::loop()
{
    _isLooping = true;
    while (_isLooping) {
        waitEpollFd();
    }
}

void SubReactor::unloop()
{
    _isLooping = false;
    wakeup();
}

void SubReactor::addConnection(int connfd)
{
    auto addConnTask = [this, connfd]() {
        addEpollReadFd(connfd);
        
        TcpConnectionPtr conn(new TcpConnection(connfd, this));
        conn->setNewConnectionCallback(_onNewConnection);
        conn->setMessageCallback(_onMessage);
        conn->setCloseCallback(_onClose);
        _conns[connfd] = conn;
        
        // 关键信息：显示哪个SubReactor处理了哪个连接
        cout << "[SubReactor-" << std::this_thread::get_id() 
             << "] 接管连接 fd=" << connfd << ", 总连接数=" << _conns.size() << endl;
        
        conn->handleNewConnectionCallback();
    };
    
    runInLoop(std::move(addConnTask));
}

void SubReactor::setNewConnectionCallback(const TcpConnectionCallback &cb)
{
    _onNewConnection = cb;
}

void SubReactor::setMessageCallback(const TcpConnectionCallback &cb)
{
    _onMessage = cb;
}

void SubReactor::setCloseCallback(const TcpConnectionCallback &cb)
{
    _onClose = cb;
}

void SubReactor::runInLoop(Functor &&cb)
{
    {
        std::lock_guard<std::mutex> lg(_mutex);
        _pendings.push_back(std::move(cb));
        // cout << "[SubReactor] runInLoop: 添加任务到队列，当前队列大小=" << _pendings.size() << endl;
    }
    wakeup();
}

void SubReactor::wakeup()
{
    // cout << "[SubReactor-" << std::this_thread::get_id() << "] wakeup() called, evtfd=" << _evtfd << endl;
    uint64_t one = 1;
    ssize_t ret = write(_evtfd, &one, sizeof(uint64_t));
    if (ret != sizeof(uint64_t)) {
        cerr << "[SubReactor] wakeup FAILED: ret=" << ret << ", errno=" << errno << endl;
        return;
    }
    // cout << "[SubReactor-" << std::this_thread::get_id() << "] wakeup() SUCCESS: wrote " << ret << " bytes" << endl;
}

int SubReactor::createEpollFd()
{
    int epfd = epoll_create1(0);
    if (epfd == -1) {
        perror("epoll_create1");
    }
    return epfd;
}

int SubReactor::createEventFd()
{
    int evtfd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
    if (evtfd == -1) {
        cerr << "[SubReactor] createEventFd FAILED: errno=" << errno << endl;
    } else {
        // cout << "[SubReactor] createEventFd SUCCESS: fd=" << evtfd << " (with EFD_CLOEXEC|EFD_NONBLOCK)" << endl;
    }
    return evtfd;
}

void SubReactor::addEpollReadFd(int fd)
{
    struct epoll_event evt;
    evt.data.fd = fd;
    evt.events = EPOLLIN | EPOLLET;
    
    // cout << "[SubReactor] addEpollReadFd: 添加fd=" << fd << " 到epoll=" << _epfd << endl;
    int ret = epoll_ctl(_epfd, EPOLL_CTL_ADD, fd, &evt);
    if (-1 == ret) {
        cerr << "[SubReactor] addEpollReadFd FAILED: fd=" << fd << ", errno=" << errno << endl;
        return;
    }
    // cout << "[SubReactor] addEpollReadFd SUCCESS: fd=" << fd << endl;
}

void SubReactor::delEpollReadFd(int fd)
{
    struct epoll_event evt;
    evt.data.fd = fd;
    evt.events = EPOLLIN;
    
    int ret = epoll_ctl(_epfd, EPOLL_CTL_DEL, fd, &evt);
    if (-1 == ret) {
        cerr << "delEpollReadFd" << endl;
        return;
    }
}

void SubReactor::waitEpollFd()
{
    int nready = 0;
    do {
        nready = epoll_wait(_epfd, _evtList.data(), _evtList.size(), 2000);
    } while (nready == -1 && errno == EINTR);
    
    if (nready == -1) {
        cerr << ">>>SubReactor TIMEOUT" << endl;
        return;
    } else if (nready == 0) {
        cout << ">>SubReactor_epoll_wait TIMEOUT." << endl;
    } else {
        if (nready == (int)_evtList.size()) {
            _evtList.resize(2 * nready);
        }
        
        for (int i = 0; i < nready; i++) {
            int fd = _evtList[i].data.fd;
            // cout << "[SubReactor] epoll事件: fd=" << fd << ", events=" << _evtList[i].events << endl;
            if (fd == _evtfd) {
                cout << "[SubReactor] 检测到eventfd事件，即将处理pending任务" << endl;
                if (_evtList[i].events & EPOLLIN) {
                    handleRead();
                    doPengdingFunctors();
                }
            } else {
                cout << "[SubReactor] 检测到连接fd=" << fd << "事件" << endl;
                if (_evtList[i].events & EPOLLIN) {
                    handleMessage(fd);
                }
            }
        }
    }
    
    // 无论是否有eventfd事件，都检查并执行pending任务
    // 这样确保定时器任务能被及时执行，即使eventfd唤醒失败
//     {
//         std::lock_guard<std::mutex> lg(_mutex);
//         if (!_pendings.empty()) {
//             cout << "[SubReactor] waitEpollFd结束时发现有 " << _pendings.size() << " 个待执行任务" << endl;
//         }
//     }
//     doPengdingFunctors();
}

void SubReactor::handleMessage(int fd)
{
    auto it = _conns.find(fd);
    if (it != _conns.end()) {
        bool isclose = it->second->isClose();
        if (!isclose) {
            // 关键信息：显示消息处理的分布（可选）
            cout << "[SubReactor-" << std::this_thread::get_id() 
                 << "] 处理消息 fd=" << fd << endl;
            it->second->handleMessageCallback();
        } else {
            cout << "[SubReactor-" << std::this_thread::get_id() 
                 << "] 连接关闭 fd=" << fd << ", 剩余连接数=" << (_conns.size() - 1) << endl;
            it->second->handleCloseCallback();
            delEpollReadFd(fd);
            _conns.erase(it);
        }
    } else {
        cerr << "TcpConnection not exist." << endl;
        return;
    }
}

void SubReactor::handleRead()
{
    uint64_t one = 1;
    // cout << "[SubReactor-" << std::this_thread::get_id() << "] handleRead() 开始读取eventfd=" << _evtfd << endl;
    ssize_t ret = read(_evtfd, &one, sizeof(uint64_t));
    if (ret != sizeof(uint64_t)) {
        cerr << "[SubReactor] handleRead FAILED: ret=" << ret << ", errno=" << errno << endl;
        return;
    }
    // cout << "[SubReactor-" << std::this_thread::get_id() << "] handleRead() SUCCESS: read " << ret << " bytes, value=" << one << endl;
}

void SubReactor::doPengdingFunctors()
{
    std::vector<Functor> tmp;
    {
        std::lock_guard<std::mutex> lg(_mutex);
        tmp.swap(_pendings);
    }
    
    cout << "[SubReactor] doPengdingFunctors: 执行 " << tmp.size() << " 个任务" << endl;
    for (auto &cb : tmp) {
        cb();
    }
}
